package org.dbstar.scheduler.concurrent;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.job.JobRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 管理多个ExecutorService
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-5-8
 */
public class MultiExecutorService extends AbstractExecutorService {
	private static final Logger LOGGER = LoggerFactory.getLogger(MultiExecutorService.class);

	private final Map<String, ExecutorService> executors;
	private final Map<String, String> typeMap;
	private final String defaultServiceKey;

	public MultiExecutorService(Map<String, ExecutorService> executors, Map<String, String> typeMap) {
		super();
		// 检查services参数的正确性
		if (executors == null || executors.isEmpty()) {
			throw new IllegalArgumentException("executors至少得有一个。");
		}
		this.executors = executors;

		// 检查typeMap参数的正确性
		if (typeMap == null) {
			throw new IllegalArgumentException("typeMap不能为null。");
		}
		if (!typeMap.containsKey("default")) {
			throw new IllegalArgumentException("typeMap中必须指定key为default的项。");
		}
		defaultServiceKey = typeMap.get("default");
		for (String key : typeMap.keySet()) {
			String val = typeMap.get(key);
			if (!executors.containsKey(val)) {
				throw new IllegalArgumentException("typeMap中key为 " + key + " 的项，对应的Executor: " + val + " 不存在。");
			}
		}
		this.typeMap = typeMap;
	}

	public Map<String, ExecutorService> getExecutors() {
		return Collections.unmodifiableMap(executors);
	}

	public Map<String, String> getTypeMap() {
		return Collections.unmodifiableMap(typeMap);
	}

	public String getDefaultServiceKey() {
		return defaultServiceKey;
	}

	public Set<String> getFullExecutors() {
		Set<String> set = new HashSet<String>();
		for (Entry<String, ExecutorService> entry : executors.entrySet()) {
			ThreadPoolExecutor e = (ThreadPoolExecutor) entry.getValue();
			if (e.getActiveCount() >= e.getMaximumPoolSize() && e.getQueue().size() > 1) {
				set.add(entry.getKey());
			}
		}
		return set;
	}

	public Set<String> getFullResources() {
		Set<String> keys = getFullExecutors();
		if (keys.isEmpty()) {
			return null;
		}
		Set<String> set = new HashSet<String>();
		for (Entry<String, String> entry : typeMap.entrySet()) {
			if (keys.contains(entry.getValue())) {
				set.add(entry.getKey());
			}
		}
		return set;
	}

	@Override
	public void shutdown() {
		for (ExecutorService service : executors.values()) {
			service.shutdown();
		}
	}

	@Override
	public List<Runnable> shutdownNow() {
		List<Runnable> list = new ArrayList<Runnable>();
		for (ExecutorService service : executors.values()) {
			list.addAll(service.shutdownNow());
		}
		return list;
	}

	@Override
	public boolean isShutdown() {
		// 所有service都shutdown后，才返回true
		for (ExecutorService service : executors.values()) {
			if (!service.isShutdown()) {
				return false;
			}
		}
		return true;
	}

	@Override
	public boolean isTerminated() {
		// 所有service都terminated后，才返回true
		for (ExecutorService service : executors.values()) {
			if (!service.isTerminated()) {
				return false;
			}
		}
		return true;
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		boolean result = true;
		InterruptedException exception = null;
		for (ExecutorService service : executors.values()) {
			try {
				result = service.awaitTermination(timeout, unit) && result;
			} catch (InterruptedException e) {
				exception = e;
			}
		}
		if (exception != null) {
			throw exception;
		}
		return result;
	}

	@Override
	public void execute(Runnable command) {
		String info[] = getExecutor(command);
		executors.get(info[0]).execute(command);
		LOGGER.info("Execute job use Executor[{}]: {}", info[0], info[1]);
	}

	protected String[] getExecutor(Runnable command) {
		if (JobFutureTask.class.isInstance(command)) {
			Job job = ((JobFutureTask) command).getJob();

			String serviceKey = null;
			// 先找资源对应的Executor
			String resource = job.getResource();
			if (resource != null) {
				serviceKey = typeMap.get(resource);
			}
			// 再找类型对应的线程池
			if (serviceKey == null) {
				serviceKey = typeMap.get(job.getType());
			}
			// 最后使用缺省的线程池
			if (serviceKey == null) {
				serviceKey = defaultServiceKey;
			}
			return new String[] { serviceKey, job.getIdentifier().getStringIdentifier() };
		} else {
			String serviceKey = defaultServiceKey;
			return new String[] { serviceKey, command.toString() };
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
		if (JobRunner.class.isInstance(callable)) {
			return (RunnableFuture<T>) newTaskFor((JobRunner) callable);
		} else {
			return super.newTaskFor(callable);
		}
	}

	private RunnableFuture<Job> newTaskFor(JobRunner runner) {
		return new JobFutureTask(runner);
	}
}
